package com.allwe.client.singleGroup;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * kafka 消费者配置
 *
 * @Author: AllWe
 * @Date: 2024/09/24/17:57
 */
@Slf4j
@Data
public class SingleGroupBaseConsumer {

    private String groupIdConfig;

    private String topicName;

    private KafkaConsumer<String, String> consumer;

    public SingleGroupBaseConsumer(String groupIdConfig, String topicName) {
        this.groupIdConfig = groupIdConfig;
        this.topicName = topicName;
        createConsumer();
    }

    private void createConsumer() {
        // 设置属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "39.107.66.31:9092");
        properties.put("key.deserializer", StringDeserializer.class);
        properties.put("value.deserializer", StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
//        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 关闭自动提交ACK确认

        consumer = new KafkaConsumer<>(properties);
    }

    public void poll() {
        try {
            consumer.subscribe(Collections.singleton(topicName));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                int count = 0;
                for (ConsumerRecord<String, String> record : records) {
                    count = 1;
                    System.out.println("partition:" + record.partition() + ",offset:" + record.offset() +",key:" + record.key() + ",value:" + record.value());
                }
                if (count == 1) {
                    // 消费到消息了就打印分隔线
                    System.out.println("===============================");
                }
            }
        } finally {
            consumer.close();
        }
    }
}
